package com.study.bigdata.spark.core.req

import org.apache.hadoop.hdfs.protocol.proto.DatanodeProtocolProtos.StorageBlockReportProto
import org.apache.spark.storage.StorageLevel
import org.apache.spark.{SparkConf, SparkContext}

object Spark01_Req_HotCategoryTop10_1 {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setMaster("local[*]").setAppName("HotCategoryTop10")
    val sc = new SparkContext(conf)
    // TODO 需求一：Top10热门品类\
    // TODO 问题：1.同一个RDD的重复使用 2.cogroup算子可能出现笛卡尔乘积，性能低下

    // TODO 读取文件，获取原始数据
    val fileDatas = sc.textFile("data/user_visit_action.txt")
//    fileDatas.persist(StorageLevel.MEMORY_AND_DISK)

    // TODO 统计品类的点击数量  先过滤再统计
    val clickDatas = fileDatas.filter(
      data => {
        val datas = data.split("_")
        val cid = datas(6)
        cid != "-1"
      }
    )

    val clickCntDatas = clickDatas.map(
      data => {
        val datas = data.split("_")
        val cid = datas(6)
        (cid, 1)
      }
    ).reduceByKey(_ + _)

    // TODO 统计品类的下单数量
    val orderDatas = fileDatas.filter(
      data => {
        val datas = data.split("_")
        val cid = datas(8)
        cid != "null"
      }
    )

    val orderCntDatas = orderDatas.flatMap(
      data => {
        val datas = data.split("_")
        val cid = datas(8)
        val cids = cid.split(",")
        cids.map((_, 1))
      }
    ).reduceByKey(_ + _)


    // TODO 统计品类的支付数量
    val payDatas = fileDatas.filter(
      data => {
        val datas = data.split("_")
        val cid = datas(10)
        cid != "null"
      }
    )

    val payCntDatas = payDatas.flatMap(
      data => {
        val datas = data.split("_")
        val cid = datas(10)
        val cids = cid.split(",")
        cids.map((_, 1))
      }
    ).reduceByKey(_ + _)

    // TODO 对统计结果进行排序 =>Touple(点击 下单 支付)
    val clickMapDatas = clickCntDatas.map {
      case (cid, cnt) => {
        (cid, (cnt, 0, 0))
      }
    }

    val orderMapDatas = orderCntDatas.map {
      case (cid, cnt) => {
        (cid, (0, cnt, 0))
      }
    }

    val payMapDatas = payCntDatas.map {
      case (cid, cnt) => {
        (cid, (0, 0, cnt))
      }
    }

    val unionRDD = clickMapDatas.union(orderMapDatas).union(payMapDatas)
    val reduceRDD = unionRDD.reduceByKey(
      // 将相同key放入同一个组对value合并
      // 此处的两个参数代表相同key的两个value值
      (t1, t2) => {
        (t1._1 + t2._1, t1._2 + t2._2, t1._3 + t2._3)
      }
    )
    val top10 = reduceRDD.sortBy(_._2,false).take(10)
    // TODO 将结果采集打印到控制台上
//    clickCntDatas.collect().foreach(println)
//    orderCntDatas.collect().foreach(println)
//    payCntDatas.collect().foreach(println)
    top10.foreach(println)
    sc.stop()

  }

}
